package com.bff.gaia.streaming.api.mo.nonoptimize;

import com.bff.gaia.api.java.DataSet;
import com.bff.gaia.api.java.ExecutionEnvironment;
import com.bff.gaia.api.java.utils.ParameterTool;

import java.util.Iterator;
import java.util.List;
import java.util.TreeSet;

public class TopNHDFS {
	public TopNHDFS() {
	}

	public static void main(String[] args) throws Exception {
		ParameterTool params = ParameterTool.fromArgs(args);
		String input = params.get("input", "hdfs://master:9000/dataset/task2.5_MessageOptimization/testData.txt");

		ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		env.getConfig().setGlobalJobParameters(params);

		DataSet<String> text = env.readTextFile(input);
		DataSet<Double> result = text.mapPartition(new MapPartitionerNon()).reduceGroup(new ReduceGroupNon());
		List<Double> collect = result.collect();
		TreeSet set = new TreeSet();
		Iterator iterator = collect.iterator();

		while(iterator.hasNext()) {
			Double list = (Double)iterator.next();
			set.add(list);
		}

		iterator = set.iterator();
		int i = 1;

		while(iterator.hasNext()) {
			Double num = (Double)iterator.next();
			if (i <= 167) {
				System.out.println(num);
				++i;
			}
		}

	}
}